package com.ethan.threadPool;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import lombok.extern.slf4j.Slf4j;
import qunar.tc.qtracer.QTracer;

import java.util.List;
import java.util.concurrent.Callable;

/**
 * @author ethan
 * create 2018-07-23 下午8:51
 * desc
 **/
@Slf4j
public class ConcurrentProcessor<T> {
//    private static final Logger log = LoggerFactory.getLogger(ConcurrentProcessor.class);

    private String taskName;

    public ConcurrentProcessor(String taskName) {
        this.taskName = taskName;
    }

    public List<T> fetchAllData(List<Callable<T>> tasks, ListeningExecutorService listeningExecutorService) {
        long startTime = System.currentTimeMillis();
        List<T> result = Lists.newArrayListWithCapacity(tasks.size());
        try {
            Preconditions.checkNotNull(tasks, "tasks不能为空!");
            List<ListenableFuture<T>> futureList = Lists.newArrayListWithCapacity(tasks.size());
            for (Callable callable : tasks) {
                futureList.add(submitCallable(callable, listeningExecutorService));
            }

            ListenableFuture<List<T>> listListenableFuture = Futures.successfulAsList(futureList);
            result = listListenableFuture.get();
        } catch (Exception e) {
            log.error("concurrent process {} exception", taskName, e);
//            QMonitor.recordOne("concurrent_process_" + taskName + "_exception");
        } finally {
//            QMonitor.recordQuantile("concurrent_process_" + taskName, System.currentTimeMillis() - startTime);
        }
        return result;
    }

    private  <T> ListenableFuture<T> submitCallable(Callable<T> callable, ListeningExecutorService listeningExecutorService) {
        ListenableFuture future;
        try {
            future = listeningExecutorService.submit(callable);
        } catch (Exception e) {
            log.error("submit {} exception", taskName, e);
//            QMonitor.recordOne("submit_" + taskName + "_exception");
            future = Futures.immediateFuture(null);
        } finally {
//            QMonitor.recordOne("sumit_" + taskName);
        }
        return future;
    }


}
